package com.zhang.req.keyedProcess;

import com.zhang.second.utils.ClickEvent;
import com.zhang.second.utils.ClickSource;
import com.zhang.second.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * 每隔5s输出url访问次数
 *
 * @title: 使用底层API实现滚动窗口
 * @author: zhang
 * @date: 2022/2/3 10:36
 */
public class TumblingWindowInKeyedProcess {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>() {
                    //key：窗口开始时间
                    //value：属于这个窗口的所有url
                    private MapState<Long, List<ClickEvent>> mapState;
                    //窗口大小
                    private long windowSize = 5000L;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(
                                new MapStateDescriptor<Long, List<ClickEvent>>(
                                        "map-state",
                                        Types.LONG, Types.LIST(Types.POJO(ClickEvent.class))
                                )
                        );
                    }

                    @Override
                    public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>.Context ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        //计算窗口开始时间
                        long currTS = ctx.timerService().currentProcessingTime();
                        long windowStartTime = currTS - currTS % windowSize;
                        //long windowEndTime = windowStartTime + windowSize;
                        if (!mapState.contains(windowStartTime)) {
                            List<ClickEvent> events = new ArrayList<>();
                            events.add(value);
                            //属于窗口第一条数据到来
                            mapState.put(windowStartTime, events);
                        } else {
                            mapState.get(windowStartTime).add(value);
                        }

                        ctx.timerService().registerProcessingTimeTimer(windowStartTime + windowSize - 1L);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>.OnTimerContext ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        long windowEndTime = timestamp + 1L;
                        long windowStartTime = windowEndTime - windowSize;
                        String key = ctx.getCurrentKey();
                        long count = mapState.get(windowStartTime).size();
                        out.collect(new UrlViewCountPerWindow(
                                key,
                                count,
                                windowStartTime,
                                windowEndTime
                        ));
                        mapState.remove(windowStartTime);
                    }
                })
                .print();

        env.execute();
    }
}
